package com.example.flinkcourse.lesson5.transform;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.functions.MapFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 数据转换实现
 */
public class DataTransform {
    private static final Logger LOG = LoggerFactory.getLogger(DataTransform.class);
    private static final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 创建数据转换函数
     *
     * @return MapFunction
     */
    public static MapFunction<String, String> createTransform() {
        return new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                try {
                    // 解析 JSON 数据
                    JsonNode jsonNode = objectMapper.readTree(value);
                    
                    // 获取操作类型
                    String op = jsonNode.get("op").asText();
                    
                    // 根据操作类型处理数据
                    switch (op) {
                        case "r":  // 读取
                        case "c":  // 创建
                            return handleCreateOrRead(jsonNode);
                        case "u":  // 更新
                            return handleUpdate(jsonNode);
                        case "d":  // 删除
                            return handleDelete(jsonNode);
                        default:
                            LOG.warn("未知的操作类型: {}", op);
                            return value;
                    }
                } catch (Exception e) {
                    LOG.error("数据转换失败: {}", value, e);
                    return value;
                }
            }
        };
    }
    
    /**
     * 处理创建或读取操作
     */
    private static String handleCreateOrRead(JsonNode jsonNode) throws Exception {
        // 获取 after 数据
        JsonNode after = jsonNode.get("after");
        if (after != null) {
            return objectMapper.writeValueAsString(after);
        }
        return null;
    }
    
    /**
     * 处理更新操作
     */
    private static String handleUpdate(JsonNode jsonNode) throws Exception {
        // 获取 after 数据
        JsonNode after = jsonNode.get("after");
        if (after != null) {
            return objectMapper.writeValueAsString(after);
        }
        return null;
    }
    
    /**
     * 处理删除操作
     */
    private static String handleDelete(JsonNode jsonNode) throws Exception {
        // 获取 before 数据
        JsonNode before = jsonNode.get("before");
        if (before != null) {
            return objectMapper.writeValueAsString(before);
        }
        return null;
    }
} 